package com.ch5;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class GroupServer {
  private int port = 8008; //服务器监听端口
  private ServerSocket serverSocket; //定义服务器套接字
  private ExecutorService executorService; //线程池
  private static final int POOL_SIZE = 10; //线程池大小
  // 线程安全的Set集合，存储所有在线客户端的Socket（static确保全局唯一）
  private static CopyOnWriteArraySet<Socket> members = new CopyOnWriteArraySet<>();

  public GroupServer() throws IOException {
    serverSocket = new ServerSocket(port);
    //创建线程池
    executorService = Executors.newFixedThreadPool(POOL_SIZE);
    System.out.println("群组服务器启动监听在 " + port + " 端口");
  }

  private PrintWriter getWriter(Socket socket) throws IOException {
    //获得输出流缓冲区的地址
    OutputStream socketOut = socket.getOutputStream();
    //网络流写出需要使用flush，这里在PrintWriter构造方法中直接设置为自动flush
    return new PrintWriter(
            new OutputStreamWriter(socketOut, "utf-8"), true);
  }

  private BufferedReader getReader(Socket socket) throws IOException {
    //获得输入流缓冲区的地址
    InputStream socketIn = socket.getInputStream();
    return new BufferedReader(
            new InputStreamReader(socketIn, "utf-8"));
  }

  // 群组发送方法：向所有在线客户端转发消息
  private void sendToAllMembers(String msg, String hostAddress) throws IOException {
    PrintWriter pw;
    OutputStream out;
    // 遍历所有在线客户端的Socket，逐个发送消息
    for (Socket tempSocket : members) {
      out = tempSocket.getOutputStream();
      pw = new PrintWriter(
              new OutputStreamWriter(out, "utf-8"), true);
      pw.println(hostAddress + " 发言：" + msg);
    }
  }

  //多客户版本，使用线程池处理多个客户端连接
  public void service() {
    while (true) {
      Socket socket = null;
      try {
        //程序阻塞等待，监听并等待客户发起连接
        socket = serverSocket.accept();
        // 新客户端连接后，将其Socket加入在线集合
        members.add(socket);
        //将客户端连接交给线程池处理
        executorService.execute(new Handler(socket));
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }

  public static void main(String[] args) throws IOException {
    new GroupServer().service();
  }

  class Handler implements Runnable {
    private Socket socket;

    public Handler(Socket socket) {
      this.socket = socket;
    }

    @Override
    public void run() {
      //获取当前客户端的IP地址，用于标识发言者
      String clientIp = socket.getInetAddress().getHostAddress();
      //本地服务器控制台显示客户端连接的用户信息
      System.out.println("New connection accepted： " + clientIp);
      try {
        BufferedReader br = getReader(socket);//定义字符串输入流
        PrintWriter pw = getWriter(socket);//定义字符串输出流

        //客户端正常连接成功，则发送服务器欢迎信息
        pw.println("From 服务器：欢迎加入群组聊天，当前在线人数：" + members.size());

        String msg = null;
        //此处程序阻塞，每次从输入流中读入一行字符串
        while ((msg = br.readLine()) != null) {
          //如果客户发送的消息为"bye"，就结束通信
          if (msg.trim().equalsIgnoreCase("bye")) {
            // 向当前客户端发送断开提示
            pw.println("From 服务器：服务器已断开连接，结束服务！");
            // 从在线集合中移除当前客户端的Socket
            members.remove(socket);
            // 通知其他在线客户端，当前用户离开
            sendToAllMembers("已离开群组", clientIp);
            System.out.println("客户端离开：" + clientIp + "，当前在线人数：" + members.size());
            break;//跳出循环读取
          }
          // 替换原单客户端回复，改为向所有在线客户端转发消息
          sendToAllMembers(msg, clientIp);
        }
      } catch (IOException e) {
        e.printStackTrace();
      } finally {
        try {
          // 客户端异常断开时，也需从在线集合移除Socket
          if (socket != null && members.contains(socket)) {
            members.remove(socket);
            System.out.println("客户端异常断开：" + clientIp + "，当前在线人数：" + members.size());
          }
          //关闭当前客户端的Socket连接及相关流
          if (socket != null)
            socket.close();
        } catch (IOException e) {
          e.printStackTrace();
        }
      }
    }
  }
}